package cn.admin.configcenter.registry.service.client;

import cn.admin.configcenter.registry.common.event.EventDispatcher;
import cn.admin.configcenter.registry.common.event.Events;
import cn.admin.configcenter.registry.common.info.ClientConnection;
import cn.admin.configcenter.registry.common.util.ReadonlyIterator;
import cn.admin.remoting.Connection;
import com.google.common.collect.MapMaker;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.*;
import java.util.concurrent.*;

public class ClientConnectionWorker {

    private static final Logger LOG = LoggerFactory.getLogger(ClientConnectionWorker.class);
    private static final int QUEUE_CAPACITY = 50000;
    private static final ThreadLocal<ClientConnectionWorker> thread2worker = new ThreadLocal<>();
    static final ConcurrentMap<ClientConnection, ClientConnection> allNativeClients =
            new MapMaker().weakKeys().weakValues().makeMap();
    private final String name;
    private final BlockingQueue<Runnable> queue;
    private final Map<Connection,ClientConnection> nativeClients;
    private final Map<String,ClientConnection> clusterClients;
    private static final String SEPARATOR = "_";

    ClientConnectionWorker(int mod, int total) {

        this.name = this.getClass().getName() + SEPARATOR + mod + "%" + total;
        this.queue = new ArrayBlockingQueue<>(QUEUE_CAPACITY);
        new InnerWorker(this.name).start();
        this.nativeClients = new HashMap<>();
        this.clusterClients = new HashMap<>();

    }

    void execute(Runnable task) {
        this.putTask(task);
    }

    <V>Future<V> execute(Callable<V> task) {
        FutureTask<V> futureTask = new FutureTask<>(task);
        this.putTask(futureTask);
        return futureTask;
    }

    private void putTask(Runnable task) {
        try {
            this.queue.put(task);
        } catch (InterruptedException e) {
            LOG.error(e.toString(),e);
        }
    }

    ClientConnection addNativeClient(Connection conn) {
        ClientConnection clientConn = this.nativeClients.get(conn);
        if (null == clientConn) {
            clientConn = new ClientConnection(conn);
            this.nativeClients.put(conn, clientConn);
            allNativeClients.put(clientConn, clientConn);
        }

        return clientConn;
    }

    public ClientConnection addClusterClient(String originalHostId, String origonalCsIp,
                                             String originalCsRole) {
        ClientConnection clientConn = this.findClusterClient(originalHostId, origonalCsIp);
        if (null == clientConn) {
            clientConn = new ClientConnection(originalHostId, origonalCsIp, originalCsRole);
            this.clusterClients.put(originalHostId + "_" + origonalCsIp, clientConn);
        }

        return clientConn;
    }

    ClientConnection removeClient(Connection conn) {
        ClientConnection client = this.nativeClients.remove(conn);
        allNativeClients.remove(client);
        if (null != client) {
            LOG.warn("[client-removed] " + client);
            EventDispatcher.fireEvent(new Events.ClientDisconnectedEvent(client));
        }

        return client;
    }

    public Iterator<ClientConnection> nativeClientsIterator() {
        return ReadonlyIterator.wrap(this.nativeClients.values().iterator());
    }

    public Iterator<ClientConnection> clusterClientsIterator() {
        return ReadonlyIterator.wrap(this.clusterClients.values().iterator());
    }

    private ClientConnection findClusterClient(String originalHostId, String remoteCsIp) {
        return this.clusterClients.get(originalHostId + "_" + remoteCsIp);
    }

    public static Iterator<ClientConnection> globalNativeClientsIter() {
        return ReadonlyIterator.wrap(allNativeClients.values().iterator());
    }

    ClientConnection removeClusterClient(String originalHostId, String originalCsIp) {
        ClientConnection client = this.clusterClients.remove(originalHostId +
                "_" + originalCsIp);
        if (null != client) {
            LOG.warn("[client-removed] {} from {}", new Object[]{client, originalCsIp});
            EventDispatcher.fireEvent(new Events.ClientDisconnectedEvent(client));
        }

        return client;
    }

    static ClientConnectionWorker getWorker() {
        return thread2worker.get();
    }

    public List<ClientConnection> getClusterClientsSnapshot() {
        return new LinkedList<>(this.clusterClients.values());
    }

    private int pendingTaskCount() {
        return this.queue.size();
    }

    void healthStatus(StringBuilder sb) {
        sb.append(this.name).append(", pending tasks: ").append(this.pendingTaskCount());
        sb.append(", clients: ").append(this.nativeClients.size());
        sb.append("/").append(this.clusterClients.size());
    }

    public List<ClientConnection> allConnectionSnapshot() {
        List<ClientConnection> result = new LinkedList<>();
        result.addAll(this.nativeClients.values());
        result.addAll(this.clusterClients.values());
        return result;
    }

    class InnerWorker extends Thread {

        InnerWorker(String name) {
            this.setDaemon(false);
            this.setName(name);
        }

        @Override
        public void run() {
            ClientConnectionWorker.thread2worker.set(ClientConnectionWorker.this);

            while (true) {
                try {
                Runnable task = ClientConnectionWorker.this.queue.take();
                long begin = System.currentTimeMillis();
                task.run();
                long duration = System.currentTimeMillis() - begin;
                if (duration > 1000L) {
                    ClientConnectionWorker.LOG.warn("it takes {}ms to run task {}", duration, task);
                }
            } catch (Throwable t) {
                ClientConnectionWorker.LOG.error("[client-worker-error] " + t.toString(), t);
            }
            }
        }
    }

}
